package com.bd03.streaminglearn.day0327

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object UpdateStateByKeyDemo {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(2))
    //updateStateByKey,checkpoint
    ssc.checkpoint("mycheckdir")
    val data = ssc.socketTextStream("hdp01",9999)
    val res:DStream[(String,Int)] = data.flatMap(t=>t.split(" ")).map((_,1)).reduceByKey(_+_)
    res.print()
    val udataFunction = (newValues:Seq[Int],runningCount:Option[Int])=>{
      val i = newValues.sum+runningCount.getOrElse(0)//a 11   a 1 a 1  a 2
      Some(i)
    }
    //key value类型的DStreming才能使用updateStateByKey
    val res2: DStream[(String, Int)] = res.updateStateByKey(udataFunction)
    res2.print()
    //updateStateBykey
    ssc.start()
    ssc.awaitTermination()
  }

}
